package state;

import bean.SensorReading;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.junit.Test;

public class Flink_State_TempDiff10 {

    public static class MyFlatMapRichFunc extends RichFlatMapFunction<SensorReading,String>{

        private Double maxDiff;

        public MyFlatMapRichFunc(Double maxDiff) {
            this.maxDiff = maxDiff;
        }
        //状态声明
        private ValueState<Double> tempState;
        @Override
        public void open(Configuration parameters) throws Exception {
            //状态初始化
            tempState = getRuntimeContext().getState(new ValueStateDescriptor<Double>("sensor-temp", Double.class));
        }

        @Override
        public void flatMap(SensorReading value, Collector<String> out) throws Exception {
            //获取状态值
            Double temp = tempState.value();
            //逻辑处理
            if(temp!=null&&(Math.abs(temp-value.getTemp())>10.0)){
                out.collect("近两次温差大于"+maxDiff+"度");

            }
            //状态更新
            tempState.update(value.getTemp());
        }
    }

    //TODO 利用状态编程实现当两次温差大于10时
    @Test
    public void test() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        env.readTextFile("input/sensorreading.txt")
                .map(new MapFunction<String, SensorReading>() {
                    @Override
                    public SensorReading map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new SensorReading(split[0],Long.parseLong(split[1]),Double.parseDouble(split[2]));
                    }
                })
                .keyBy(SensorReading::getId)
                .flatMap(new RichFlatMapFunction<SensorReading, String>() {
                    private ValueState<Double> diff;

                    @Override
                    public void open(Configuration configuration){
                        diff=getRuntimeContext().getState(new ValueStateDescriptor<Double>("temp",Double.class));
                    }

                    @Override
                    public void flatMap(SensorReading value, Collector<String> out) throws Exception {
                        Double temp = diff.value();
                        if(temp!=null && Math.abs(temp-value.getTemp())>10L){
                            out.collect("当前温差较大>"+Math.abs(temp-value.getTemp())+",两次温度分别为："+temp+"~"+value.getTemp());
                        }
                        diff.update(value.getTemp());
                    }
                })
                .print();
        env.execute();
    }

}
